package com.lxz.rpcbasednetty;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.UUID;

import org.jboss.netty.handler.codec.serialization.ClassResolvers;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolver;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;

public class RpcClient extends ChannelHandlerAdapter{
	private static final Logger LOGGER = LoggerFactory.getLogger(RpcClient.class);
	private final String host;
	private final int port;
	private RpcResponse response;
	private final Object obj = new Object();
	
	public RpcClient(String host, int port){
		this.host = host;
		this.port = port;
	}
	
	public RpcResponse send(RpcRequest request) throws Exception{
		//配置客户端NIO线程组
		EventLoopGroup group = new NioEventLoopGroup();
		try{
			Bootstrap b = new Bootstrap();
			b.group(group).channel(NioSocketChannel.class)
				.option(ChannelOption.SO_KEEPALIVE, true)
				.handler(new ChannelInitializer<SocketChannel>(){
					@Override
					protected void initChannel(SocketChannel ch) throws Exception {
						ch.pipeline().addLast(new ObjectDecoder((ClassResolver) ClassResolvers.cacheDisabled(this.getClass().getClassLoader())))
							.addLast(new ObjectEncoder())
							.addLast(RpcClient.this);
					}
				});
			
			//发起异步连接操作
			ChannelFuture f = b.connect(host, port).sync();
			f.channel().writeAndFlush(request).sync();
			
			synchronized(obj){
				obj.wait();
			}
			
			if(response != null){
				f.channel().closeFuture().sync();//等待客户端链路关闭
			}
			System.out.println(response.getRequestId());
			return response;
		}finally{
			//优雅退出，释放NIO线程组
			group.shutdownGracefully();
		}
	}
	
	@SuppressWarnings("unchecked")
	public <T> T create(Class<?> interfaceClass){
		return (T)Proxy.newProxyInstance(interfaceClass.getClassLoader(), 
				new Class<?>[]{interfaceClass}, 
				new InvocationHandler(){
					public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
						RpcRequest request = new RpcRequest();
						request.setRequestId(UUID.randomUUID().toString());
						request.setClassName(method.getDeclaringClass().getName());
						request.setMethodName(method.getName());
						request.setParameterTypes(method.getParameterTypes());
						request.setParameters(args);
						
						RpcClient client = new RpcClient("127.0.0.1", 1234);
						RpcResponse response = client.send(request);
						
						if(response.isError()){
							throw response.getError();
						}else{
							return response.getResult();							
						}
					}
				});
	}
		
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception{
		this.response = (RpcResponse)msg;
		synchronized(obj){
			obj.notifyAll();
		}
	}
	
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause){
		LOGGER.warn("Unexcepted exception from downstream : " + cause.getMessage());
		ctx.close();
	}

}
